package goldberg.drew.concurrent;

import java.io.IOException;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;

import java.net.URL;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;


import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* Or should I just use the equivalent of Concurrent data structure of an ArrayList and Locks?  for locks use this format
* 
* aMonitor.lock();
* try {
*
* }
* finally {
* aMonitor.unlock();
* }
*
*/

public class ConcurrentTwitter extends AbstractConcurrentTwitter{
	
	private static int poolSize;
	private static ExecutorService executorPool;	
	private static final ReadWriteLock monitor = new ReentrantReadWriteLock();
        
	/**
	* Empty Constructor
	*/
	private ConcurrentTwitter() {
	}
	
	/**
	* Creates the directory hierarchy and textfile for each user	
	* usersConcurrent/userName/userName.txt
	* intializes the threadPool executorPool
	*/
	private void init() {

		ConcurrentTwitter.createUsers();
		poolSize = calculateThreadPoolSize();
		executorPool = Executors.newFixedThreadPool(poolSize);
		runThreads();	
	}
	
	public static ConcurrentTwitter create() {
		final ConcurrentTwitter concurrentTwitter = new ConcurrentTwitter();
			concurrentTwitter.init();
		return concurrentTwitter;
	}


	public int calculateThreadPoolSize() {
		final int numberOfCores = Runtime.getRuntime().availableProcessors();
		final double blockingCoefficient = 0.9;
		final int threadPoolSize = (int) (numberOfCores/(1 - blockingCoefficient));
		System.out.println("The number of cores is " + numberOfCores);
		System.out.println("The blocking coefficient is " + blockingCoefficient);
		System.out.println("The number of threads is " + threadPoolSize);		
		return threadPoolSize;	
	}	
	

	public void runThreads() {
		for(final String key : users.keySet()) {
			executorPool.submit(new Runnable() {
				public void run(){
					getTweets(key, users.get(key));
				}								
			});
		}
		executorPool.shutdown();
		
		//must ensure that all the tasks have completed before we try to get the time it took to run.
		//executorPool.awaitTermination:   Blocks until all tasks have completed execution after a shutdown request, or the timeout occurs, or the 			current	thread is interrupted, whichever happens first. 		
		try{
			executorPool.awaitTermination(100, TimeUnit.SECONDS);	
		}
		catch(InterruptedException e) {
			e.printStackTrace();		
		}
	}
       
        
	public void getTweets(String userName, File file) {
		BufferedInputStream in = null; 
	 	BufferedOutputStream bout = null;
		FileOutputStream fos =  null;
	      	for(int i = 1; i < 4; i++) {
			try {	
				System.out.println("Retreiving User " + userName + " page " + i);	 			
				in = new BufferedInputStream(new URL("http://api.twitter.com/1/statuses/user_timeline.xml?id="+userName+"&count=200&page="+i).openStream());
				fos = new FileOutputStream(file.toString(), true);
				bout = new BufferedOutputStream(fos, 1024);
				byte[] data = new byte[1024];
				int x = 0; 
				while((x=in.read(data,0,1024))>0){
				   bout.write(data, 0, x);
		        	}
				bout.close();
				in.close();
			}
			catch (IOException e) {
				System.out.println("Failed at userName " + userName + " page " + i);
				e.printStackTrace();
			}
	        }
	    	
	 }
	public static void main(String[] args) {
	
		ConcurrentTwitter ct = new ConcurrentTwitter();
		final long startTime = System.nanoTime();		
		ct.create();
		final long finishTime = System.nanoTime();
		if(ct.executorPool.isTerminated())
			System.out.println("If all the tasks have been completed properly.\nTime take in seconds " + (finishTime - startTime / 1.0e9));
		else{
			System.out.println("This was printed because the executor pool threads haven't completed yet.\nTime take in seconds " + (finishTime - startTime / 1.0e9));
		}
        }
	






}
